package com.hucais.utils;

import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 共用的APP工具
 */
public class AppCommonUtil {

    /**
     * 获取默认的执行环境
     *
     * @param obsPathSuf        checkpoint路径的后缀
     * @param sourceParallelism 源端的并行度
     * @return Flink默认的执行环境
     */
    public static StreamExecutionEnvironment getDefaultStreamEnv(boolean isLocal,String obsPathSuf, int sourceParallelism) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(sourceParallelism);
        if(!isLocal){
            // 设置checkPoint
            String checkPointPath = DefaultPropertiesUtil.get("obs.path.pre") + obsPathSuf;
            RocksDBStateBackend rocksDbBackend = new RocksDBStateBackend(new FsStateBackend(checkPointPath), true);
            env.setStateBackend(rocksDbBackend);
            // 开启Flink CheckPoint配置，开启时若触发CheckPoint，会将Offset信息同步到Kafka
            env.enableCheckpointing(30000);
            // 设置精确一次模式
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            // 设置两次checkpoint的最小间隔时间
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(6000);
            // 设置checkpoint超时时间
            env.getCheckpointConfig().setCheckpointTimeout(60000);
            // 设置checkpoint最大并发数
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
            // 设置作业取消时保留checkpoint
            env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        }
        return env;
    }

}
